package mysql.socket;

import io.netty.buffer.ByteBuf;
import mysql.MysqlPacketHeader;
import mysql.cs.ComAndPackManager;
import mysql.cs.PacketManager;
import mysql.cs.handshake.HandshakeV10;
import mysql.cs.packet.ErrorPacket;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @ClassName MysqlNettyConnect
 * @Description TODO
 * @Author fengk
 * @Date 2018/9/4 16:37
 */

public class MysqlNettyConnect {
    private static final Logger logger  = LoggerFactory.getLogger(MysqlNettyConnect.class);
    private InetSocketAddress address;
    private String userName;
    private String password;
    private SocketChannel channel;
    private long connectid=-1;


    private int                 soTimeout         = 30 * 1000;
    private int                 connTimeout       = 5 * 1000;
    private int                 receiveBufferSize = 16 * 1024;
    private int                 sendBufferSize    = 16 * 1024;


    private volatile boolean dumping =false;
    public static final int     timeout           = 5 * 1000; //5s
    /**
     * Number
     *
     * Hex
     *
     * Character Set Name
     *
     * 8
     *
     * 0x08
     *
     * latin1_swedish_ci
     *
     * 33
     *
     * 0x21
     *
     * utf8_general_ci
     *
     * 63
     *
     * 0x3f
     *
     * binary
     */
    private byte charsetNumber=33;
    private String defaultSchema="test";

    private AtomicBoolean connected=new AtomicBoolean(false);

    public MysqlNettyConnect(InetSocketAddress address, String userName, String password) {
        String addrStr=address.getHostString();
        int port=address.getPort();

        this.address = new InetSocketAddress(addrStr,port);
        this.userName = userName;
        this.password = password;
    }

    private MysqlNettyConnect(){

    }
    public MysqlNettyConnect(InetSocketAddress address, String userName, String password, byte charsetNumber, String defaultSchema) {
        this(address,userName,password);
        this.charsetNumber = charsetNumber;
        this.defaultSchema = defaultSchema;
    }
    public MysqlNettyConnect connect() throws Exception {
        if(connected.compareAndSet(false,true)){
            try{
                channel=NettyCachePool.open(address);
                logger.info("connect mysql to {}...", address);
                login(channel);
            }catch (Exception e){
                disconnect();
                throw  new IOException("connect "+this.address+" failure",e);
            }

        }else {
            logger.error("the channel can not be connected twice");
        }
        return this;
    }

    public MysqlNettyConnect dumpBinLog(String binlogFile,long serverid,long position) throws IOException {
        byte[] binlogDumpCom=ComAndPackManager.binlogDumpCommandPacketByte(binlogFile,serverid,position);
        PacketManager.writeBody(channel,binlogDumpCom);
        return this;
    }

    public void reconnect() throws Exception {
        disconnect();
        connect();
    }
    private void disconnect() throws IOException {
        if(connected.compareAndSet(true,false)){
            try{
                if(channel!=null){
                    channel.close();
                }
                logger.info("disconnect mysql to {} ...",address);
            }catch (Exception e){
                throw  new IOException("disconnect "+this.address +" failure",e);
            }

            if(dumping && connectid>0){
                MysqlNettyConnect connect=null;
                try {
                    connect=this.fork();
                    connect.connect();;
                    ComAndPackManager.queryCommandPacketByte("KILL CONNECTION "+connectid);
                }catch (Exception e){
                    //
                    logger.info("KILL DUMP " + connectid + " failure", e);

                }finally {
                    if(connect!=null){
                        connect.disconnect();
                    }
                }
                dumping=false;
            }
        }else {
            logger.info("the channel {} is not connected", this.address);
        }
    }

    private void login(SocketChannel channel) throws IOException {
        MysqlPacketHeader header=PacketManager.readHeader(channel,4,timeout);
        byte[] body=PacketManager.readBytes(channel,header.getPayloadLength(),timeout);
        if(body[0]==(byte)0x0a){
            //登录信息
            ComAndPackManager comAndPackManager=new ComAndPackManager(userName,password);
            comAndPackManager.HandshakeV10EncodeAndSetRES(body);
            HandshakeV10 handshakeV10=comAndPackManager.getHandshakeV10();
            connectid=handshakeV10.connectId;
            PacketManager.writeBodyNext(channel,comAndPackManager.getHandshakeResponse41().decode());
            logger.info("client authentication packet is sent out.");

        }else if(body[0]==(byte)0xff){
            ErrorPacket errorPacket=ComAndPackManager.ErrorPacketEncode(body);
            throw new IOException("handshake exception:\n"+errorPacket.message);
        }else if(body[0]==(byte)0x00){
            throw new IOException("Unexpected OK packet at handshake phase.");
        }else if(body[0]==(byte)0xfe){
            throw new IOException("Unexpected EOF packet at handshake phase.");
        }else {
            throw new IOException("Unexpected Other(SSL?) packet at handshake phase.");
        }


        // check auth result
        header = null;
        header = PacketManager.readHeader(channel, 4);
        body = null;
        body = PacketManager.readBytes(channel, header.getPayloadLength(), timeout);
        assert body != null;

        if(body[0]==(byte)0xff){
            ErrorPacket errorPacket=ComAndPackManager.ErrorPacketEncode(body);
            throw new IOException("handshake exception:\n"+errorPacket.message);
        }else if(body[0]==(byte)0x00){
            logger.info("login success ...");
        }else if(body[0]==(byte)0xfe){
            throw new IOException("Unexpected EOF packet at handshake phase.");
        }else {
            throw new IOException("Unexpected Other(SSL?) packet at handshake phase.");
        }
    }

    public void quit() throws IOException {
        PacketManager.writePkg(channel, ComAndPackManager.quitCommandPacketDecode());

    }
    public boolean isConnected() {
        return this.channel != null && this.channel.isConnected();
    }

    public MysqlNettyConnect fork() {
        MysqlNettyConnect connector = new MysqlNettyConnect();
        connector.setCharsetNumber(getCharsetNumber());
        connector.setDefaultSchema(getDefaultSchema());
        connector.setAddress(getAddress());
        connector.setPassword(password);
        connector.setUserName(getUserName());
        connector.setReceiveBufferSize(getReceiveBufferSize());
        connector.setSendBufferSize(getSendBufferSize());
        connector.setSoTimeout(getTimeout());
        connector.setConnTimeout(connTimeout);
        return connector;
    }

    public InetSocketAddress getAddress() {
        return address;
    }

    public void setAddress(InetSocketAddress address) {
        this.address = address;
    }

    public String getUserName() {
        return userName;
    }

    public void setUserName(String userName) {
        this.userName = userName;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public SocketChannel getChannel() {
        return channel;
    }

    public void setChannel(SocketChannel channel) {
        this.channel = channel;
    }

    public long getConnectid() {
        return connectid;
    }

    public void setConnectid(long connectid) {
        this.connectid = connectid;
    }

    public static int getTimeout() {
        return timeout;
    }

    public byte getCharsetNumber() {
        return charsetNumber;
    }

    public void setCharsetNumber(byte charsetNumber) {
        this.charsetNumber = charsetNumber;
    }

    public String getDefaultSchema() {
        return defaultSchema;
    }

    public void setDefaultSchema(String defaultSchema) {
        this.defaultSchema = defaultSchema;
    }

    public int getSoTimeout() {
        return soTimeout;
    }

    public void setSoTimeout(int soTimeout) {
        this.soTimeout = soTimeout;
    }

    public int getConnTimeout() {
        return connTimeout;
    }

    public void setConnTimeout(int connTimeout) {
        this.connTimeout = connTimeout;
    }

    public int getReceiveBufferSize() {
        return receiveBufferSize;
    }

    public void setReceiveBufferSize(int receiveBufferSize) {
        this.receiveBufferSize = receiveBufferSize;
    }

    public int getSendBufferSize() {
        return sendBufferSize;
    }

    public void setSendBufferSize(int sendBufferSize) {
        this.sendBufferSize = sendBufferSize;
    }

    public boolean isDumping() {
        return dumping;
    }

    public void setDumping(boolean dumping) {
        this.dumping = dumping;
    }
}
